package edu.zju.jzs.firstsort.process;


import edu.zju.jzs.firstsort.entity.FlowBean;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * Created by 11544 on 2017/10/13.
 */
public class Main {
    public static void main(String[] args) throws Exception
    {
        Configuration conf=new Configuration();
        Job job=Job.getInstance(conf);

        job.setMapperClass(FlowCountMapper.class);
        job.setReducerClass(FlowCountReducer.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(FlowBean.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(FlowBean.class);

        String inputfile="D:\\hadoopdata\\data\\phoneflow.txt";
        Path inpath=new Path(inputfile);
        SimpleDateFormat sdf=new SimpleDateFormat("yyyy-MM-dd-HH-mm-ss-SSS__2_");
        String outputfilename=sdf.format(new Date())+"flowcount";

        String outpufile="D:\\hadoopdata\\"+outputfilename;
        Path outpath=new Path(outpufile);

        FileInputFormat.setInputPaths(job,inpath);
        FileOutputFormat.setOutputPath(job,outpath);

        job.waitForCompletion(true);

    }
}
